package cn.doitedu;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.List;
import java.util.Map;

public class TimestampFlumeInterceptor implements Interceptor {
    String headerKeyName;
    String extractTimeName;

    private TimestampFlumeInterceptor(String headerKeyName,String extractTimeName){
        this.headerKeyName = headerKeyName;
        this.extractTimeName = extractTimeName;
    }


    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
        byte[] bodyBytes = event.getBody();
        String logJson = new String(bodyBytes);

        // 抽行为时间
        JSONObject jsonObject = JSON.parseObject(logJson);

        Long actionTime = jsonObject.getLong(extractTimeName);


        Map<String, String> headers = event.getHeaders();
        headers.put(headerKeyName,actionTime+"");

        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {

        for (Event event : list) {
            intercept(event);
        }

        return list;
    }

    @Override
    public void close() {


    }

    public static class TimestampFlumeInterceptorBuilder implements Interceptor.Builder{

        String headerKeyName;
        String extractTimeName;

        @Override
        public Interceptor build() {

            return new TimestampFlumeInterceptor(this.headerKeyName,this.extractTimeName);
        }

        @Override
        public void configure(Context context) {

            //a1.sources.s1.interceptors.i1.header_key_name = timestamp
            //a1.sources.s1.interceptors.i1.extract_time_name = action_time

            this.headerKeyName = context.getString("header_key_name");
            this.extractTimeName = context.getString("extract_time_name");

            System.out.println("拦截器的configure被调用了，拿到的参数是: headerKeyName=>" + headerKeyName +" , extractTimeName=>" + extractTimeName);

        }
    }


}
